﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.IO;
using System.Threading.Tasks;
using System.Threading;

namespace IGolovkine.AoA
{
    public class AsyncStream : IChannel<byte[]>, IObservable<byte[]>, IObserver<byte[]>, IDisposable
    {
        private readonly Stream _stream;
        private readonly int _bufferSize;
        private IObserver<byte[]> _observer;
        private volatile bool _cancelled;

        public AsyncStream(Stream stream, int bufferSize)
        {
            _stream = stream;
            _bufferSize = bufferSize;
        }

        public IDisposable Subscribe(IObserver<byte[]> observer)
        {
            _observer = observer;
            Task task = Task.Factory.StartNew(() =>
            {
                byte[] buffer = new byte[_bufferSize];
                int lenght = 0;
                try
                {
                    while (_cancelled == false && (lenght = _stream.Read(buffer, 0, buffer.Length)) != 0)
                    {
                        byte[] copy = new byte[lenght];
                        Array.Copy(buffer, copy, lenght);
                        observer.OnNext(copy);
                    }
                }
                catch (Exception ex)
                {
                    observer.OnError(ex);
                }
                finally
                {
                    observer.OnCompleted();
                }
            });

            return Disposable.Create(() =>
            {
                this.Dispose();
                task.Wait();
            });
        }

        public IObservable<byte[]> Output
        {
            get { return this; }
        }

        public IObserver<byte[]> Input
        {
            get { return this; }
        }

        void IObserver<byte[]>.OnCompleted()
        {
            this.Dispose();
        }

        void IObserver<byte[]>.OnError(Exception exception)
        {
            if (_observer != null)
            {
                _observer.OnError(exception);
            }
            else
            {
                throw exception;
            }
        }

        void IObserver<byte[]>.OnNext(byte[] value)
        {
            try
            {
                _stream.Write(value, 0, value.Length);
                _stream.Flush();
            }
            catch 
            {
                this.Dispose();
            }
        }

        public void Dispose()
        {
            _cancelled = true;
            _observer = null;
            _stream.Dispose();

        }
    }
}
